package com.cennavi.funcs;

import java.text.DecimalFormat;

import org.apache.spark.api.java.function.PairFunction;

import com.google.common.base.Optional;

import scala.Tuple2;

/** * @author  作者 :allen
    * @date 创建时间：2017年10月10日 上午10:03:18 
    * @version 1.0 
    */

public class Join2RticOutputPairFunction
		implements PairFunction<Tuple2<String, Tuple2<Optional<String>, Optional<String>>>, String, String> {
	private static final long serialVersionUID = 1L;

	public Tuple2<String, String> call(Tuple2<String, Tuple2<Optional<String>, Optional<String>>> arg0)
			throws Exception {
		//rticid
		String key = arg0._1();
		//检测器结果
		String value1 = "";
		//反推结果
		String value2 = "";
		//合并后结果
		String value = "";
		String flow = "";
		String saturation = "";
		String capacity = "";
		String flow1 = "null";
		String flow2 = "null";
		if(arg0._2()._1().isPresent()){
			value1 = arg0._2()._1().get();
		}
		if(arg0._2()._2().isPresent()){
			value2 = arg0._2()._2().get();
		}
		//三种情况，都不是空、检测器不是空而反推是空、反推不是空而检测器是空
		if(!value1.equals("") && !value2.equals("")){
			flow1 = value1.split(",")[8];
			flow2 = value2.split(",")[8];
			//流量取二者的加权平均
			flow = String.valueOf((int)Math.ceil(Double.parseDouble(flow1) * 0.5 + Double.parseDouble(flow2) * 0.5));
			capacity = value2.split(",")[10];
			//饱和度重新计算
			saturation = new DecimalFormat("######0.00").format(Double.parseDouble(flow) / Double.parseDouble(capacity));
			value = value2.split(",")[0] + "," + 
					value2.split(",")[1] + "," +
					value2.split(",")[2] + "," +
					value2.split(",")[3] + "," + 
					"3," +
					value2.split(",")[5] + "," + 
					value2.split(",")[6] + "," + 
					value2.split(",")[7] + "," + 
					flow + "," +
					saturation + "," +
					capacity + "," + 
					flow1 + "," +
					flow2;
		}else if(!value1.equals("")){
			flow1 = value1.split(",")[8];
			//value = value1 + "," + flow1 + "," + flow2;
			value = value1.split(",")[0] + "," + 
					value1.split(",")[1] + "," +
					value1.split(",")[2] + "," +
					value1.split(",")[3] + "," + 
					"3," +
					value1.split(",")[5] + "," + 
					value1.split(",")[6] + "," + 
					value1.split(",")[7] + "," + 
					flow1 + "," +
					value1.split(",")[9] + "," +
					value1.split(",")[10] + "," +
					flow1 + "," +
					flow2;
		}else{
			flow2 = value2.split(",")[8];
			//value = value2 + "," + flow1 + "," + flow2;
			value = value2.split(",")[0] + "," + 
					value2.split(",")[1] + "," +
					value2.split(",")[2] + "," +
					value2.split(",")[3] + "," + 
					"3," +
					value2.split(",")[5] + "," + 
					value2.split(",")[6] + "," + 
					value2.split(",")[7] + "," + 
					flow2 + "," +
					value2.split(",")[9] + "," +
					value2.split(",")[10] + "," +
					flow1 + "," +
					flow2;
		}
		return new Tuple2<String,String>(key, value);
	}

}
